package main.niit

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

import java.util.Properties

//最近20年(1998年--2018年)不同品牌二手车辆销售数量。
object salesVolume {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.OFF) //省略运行日志

    val spark =SparkSession.builder()
      .appName("salesVolume")
      .master("local[*]")
      .getOrCreate()

    val jdbcDF =spark.read.format("jdbc") //从数据库中导入本地
      .option("url","jdbc:mysql://192.168.56.104:3306/cardata?useSSL=false")
      .option("driver","com.mysql.jdbc.Driver")
      .option("dbtable","data")
      .option("user","root")
      .option("password","root")
      .load()

    import spark.implicits._

    val prop =new Properties()
    prop.put("user","root")
    prop.put("password","root")
    prop.put("driver","com.mysql.jdbc.Driver")

    val filterDF=jdbcDF.select("汽车品牌","注册年份").filter($"注册年份">1997).filter($"注册年份"<2019)
    val resultDF=filterDF.withColumn("数量",lit(1))
      .groupBy("汽车品牌")
      .agg(sum("数量").alias("总车辆数"))

    resultDF.write.mode("append")
      .jdbc("jdbc:mysql://192.168.56.104:3306/cardata?useSSL=false","cardata.salesVolume",prop)






  }

}
